package com.dja.nzw.mq.rabbit.consumer;

import com.dja.nzw.mq.rabbit.EndPoint;
import com.dja.nzw.mq.rabbit.config.AMQPConfig;
import com.dja.nzw.mq.rabbit.dto.MessagePayload;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by shuyun on 2016/8/9.
 */
public class Consumer extends EndPoint {
    private AmqpAdmin amqpAdmin;

    public Consumer(AMQPConfig amqpConfig) {
        super(amqpConfig);
        amqpAdmin  = new RabbitAdmin(connectionFactory);
    }

    public Consumer dataBinding() {
        //declare queue
        Queue queue = queue();
        amqpAdmin.declareQueue(queue);

        //declare exchange
        Exchange exchange = exchange();
        amqpAdmin.declareExchange(exchange);

        //data binding
        Binding binding = new BindingBuilder().bind(queue).to(exchange).with("seckill").and(null);
        amqpAdmin.declareBinding(binding);

        return this;
    }

    private Queue queue(){
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("x-ha-policy", "all");
        Queue queue = new Queue(MessagePayload.queueName(), true, false, false, map);
        return queue;
    }

    private Exchange exchange(){
        Exchange exchange = new DirectExchange(MessagePayload.exchangeName(), true, false, null);
        return exchange;
    }

    public void startListener(MessageListener messageListener) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setAutoStartup(false);
        container.setDeclarationRetries(100000000);
        container.setRecoveryInterval(30000);
        container.setQueues(queue());
        container.setMessageListener(messageListener);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);

        container.start();
    }
}
